﻿/*======================================================================
== Copyright : BlueCurve (c)
== Licence   : Gnu/GPL v2.x
== Author    : Teddy Albina
== Email     : bluecurveteam@gmail.com
== Web site  : http://www.codeplex.com/BlueCurve
========================================================================*/
using System;
using System.Collections.Generic;
using System.Security;
using System.Security.Permissions;
using System.Threading;

namespace BlueCurve.Common.Threads
{
    /// <summary>
    /// Implémentation d'un pool de threads
    /// </summary>
    public class SimpleLockThreadPool : PermissionSet, IThreadPool
    {
        #region 'Configuration'
        
        // Constructors--
        // Two things may be specified:
        //   ConcurrencyLevel == fixed # of threads to use
        //   FlowExecutionContext == whether to capture & flow ExecutionContexts for work items
        /// <summary>
        /// Constructeur
        /// </summary>
        public SimpleLockThreadPool()
            : base(PermissionState.None)
        {
            this.m_concurrencyLevel = Environment.ProcessorCount * 15;
            this.m_flowExecutionContext = true;
            SetPermission();
        }

        /// <summary>
        /// Constructeur
        /// </summary>
        /// <param name="concurrencyLevel">Nombre de thread maximal dans le pool</param>
        public SimpleLockThreadPool(int concurrencyLevel)
            : base(PermissionState.None)
        {
            this.m_concurrencyLevel = concurrencyLevel;
            this.m_flowExecutionContext = true;
            SetPermission();
        }

        /// <summary>
        /// Constructeur
        /// </summary>
        /// <param name="flowExecutionContext">Permet la capture du flux de travail,et du contexte d'exécution si la valeur est "true"</param>
        public SimpleLockThreadPool(bool flowExecutionContext)
            : base(PermissionState.None)
        {
            this.m_concurrencyLevel = Environment.ProcessorCount;
            this.m_flowExecutionContext = flowExecutionContext;
            SetPermission();
        }

        /// <summary>
        /// Constructeur
        /// </summary>
        /// <param name="concurrencyLevel">Nombre de thread maximal dans le pool</param>
        /// <param name="flowExecutionContext">Permet la capture du flux de travail,et du contexte d'exécution si la valeur est "true"</param>
        public SimpleLockThreadPool(int concurrencyLevel, bool flowExecutionContext) :
            base(PermissionState.None)
        {
            if (concurrencyLevel <= 0)
                throw new ArgumentOutOfRangeException("concurrencyLevel");

            m_concurrencyLevel = concurrencyLevel;
            m_flowExecutionContext = flowExecutionContext;

            // If suppressing flow, we need to demand permissions.
            if (!flowExecutionContext)
                new SecurityPermission(SecurityPermissionFlag.Infrastructure).Demand();

            SetPermission();
        }

        /// <summary>
        /// Attribut des permissions au pool de threads
        /// </summary>
        private void SetPermission()
        {
            base.AddPermission(new SecurityPermission(SecurityPermissionFlag.Execution));
        }



        #endregion

        // Each work item consists of a closure: 
        /// <summary>
        /// Chaque flux de travail à une closure: work + (optional) state obj + context.
        /// </summary>
        internal struct WorkItem
        {
            /// <summary>
            /// Fonction attaché au thread
            /// </summary>
            internal WaitCallback m_work;
            /// <summary>
            /// Objet de travail du thread
            /// </summary>
            internal object m_obj;
            /// <summary>
            /// Contexte d'exécution du thread
            /// </summary>
            internal ExecutionContext m_executionContext;
                      
            /// <summary>
            /// Flux de travail
            /// </summary>
            /// <param name="work">Fonction attaché au thread</param>
            /// <param name="obj">Objet de travail</param>
            internal WorkItem(WaitCallback work, object obj)
            {
                m_work = work;
                m_obj = obj;
                m_executionContext = null;
            }
                        
            /// <summary>
            /// Méthode d'invocation du thread
            /// </summary>
            internal void Invoke()
            {
                // Execution normale (delegate invoke) ou sous le contexte approprié
                if (m_executionContext == null)
                    m_work(m_obj);
                else
                    ExecutionContext.Run(m_executionContext, ContextInvoke, null);
            }                        

            /// <summary>
            /// Attribut le contexte d'invocation
            /// </summary>
            /// <param name="obj">Objet de travail</param>
            private void ContextInvoke(object obj)
            {
                m_work(m_obj);
            }
        }
        

        /// <summary>
        /// Nombre de thread en cours
        /// </summary>
        private long m_runningThreads = 0;
        /// <summary>
        /// Nombre maximal de thread dans le pool
        /// </summary>
        private readonly int m_concurrencyLevel;
        /// <summary>
        /// Autorisation d'interception du flux de travail
        /// et du contexte d'execution
        /// </summary>
        private readonly bool m_flowExecutionContext;
        /// <summary>
        /// Pile des work item
        /// </summary>
        private readonly Queue<WorkItem> m_queue = new Queue<WorkItem>();
        /// <summary>
        /// Tableau des threads
        /// </summary>
        private Thread[] m_threads;
        /// <summary>
        /// Nombre de threads en attente
        /// </summary>
        private int m_threadsWaiting;
        /// <summary>
        /// Arrete le pool
        /// </summary>
        private bool m_shutdown;
        
        // Methods to queue work.
        /// <summary>
        /// Méthode de travaille
        /// </summary>
        /// <param name="work">Méthode à executer</param>
        public void QueueUserWorkItem(WaitCallback work)
        {
            QueueUserWorkItem(work, null);
        }


        /// <summary>
        /// Méthode de travaille
        /// </summary>
        /// <param name="work">Méthode à executer</param>
        /// <param name="obj">Objet de travail</param>
        public void QueueUserWorkItem(WaitCallback work, object obj)
        {
            WorkItem wi = new WorkItem(work, obj);

            // Si le flux d'execution est sur true, capture du contexte d'appel
            if (m_flowExecutionContext)
                wi.m_executionContext = ExecutionContext.Capture();

            // Permet d'être sure que le pool à bien été démarré
            EnsureStarted();
            
            // Insertion du work item dans la pile
            lock (m_queue)
            {
                m_queue.Enqueue(wi);
                if (m_threadsWaiting > 0)
                    Monitor.Pulse(m_queue);
            }
        }


        /// <summary>
        /// Retourne le nombre de threads en cours de fonctionnement
        /// </summary>
        public long CurrentThreads
        {
            get { return Interlocked.Read(ref m_runningThreads); }
        }


        /// <summary>
        /// Retourne le nombre maximal de thread authorisé
        /// </summary>
        public int MaxThreads
        {
            get { return m_concurrencyLevel; }
        }


        /// <summary>
        /// S'assure que les threas ont été lancés
        /// </summary>
        private void EnsureStarted()
        {
            if (m_threads == null)
            {
                lock (m_queue)
                {
                    if (m_threads == null)
                    {
                        m_threads = new Thread[m_concurrencyLevel];
                        for (int i = 0; i < m_threads.Length; i++)
                        {
                            m_threads[i] = new Thread(DispatchLoop);
                            m_threads[i].Start();
                            Interlocked.Increment(ref m_runningThreads);
                        }
                    }
                }
            }
        }


        /// <summary>
        /// Fonction qui répartit les threads
        /// </summary>
        private void DispatchLoop()
        {
            while (true)
            {
                WorkItem wi = default(WorkItem);
                lock (m_queue)
                {
                    // On termine le thread si shutdown à été demandé
                    if (m_shutdown)
                        return;

                    // Recherche un nouveau work item à executer
                    while (m_queue.Count == 0)
                    {
                        m_threadsWaiting++;
                        try { Monitor.Wait(m_queue); }
                        finally { m_threadsWaiting--; }

                        if (m_shutdown)
                            return;
                    }
                    wi = m_queue.Dequeue();
                }

                // Invocation.
                wi.Invoke();
            }
        }


        /// <summary>
        /// Dispose signal l'arret du pool, et attend que tout les threads soit fini
        /// </summary>
        public void Dispose()
        {
            m_shutdown = true;
            lock (m_queue)
            {
                Monitor.PulseAll(m_queue);
            }

            for (int i = 0; i < m_threads.Length; i++)
            {
                m_threads[i].Join();
                Interlocked.Decrement(ref m_runningThreads);
            }
        }
    }

    // =========== Exemple d'utilisation
    /*
    class Program
    {

        public static void Main(string[] args)
        {

            bool separateQueueFromDrain = bool.Parse(args[0]);

            const int warmupRunsPerThreadPool = 100;

            const int realRunsPerThreadPool = 1000000;



            IThreadPool[] threadPools = new IThreadPool[]

        {

            new CLRThreadPool(),

            new CLRUnsafeThreadPool(),

            new SimpleLockThreadPool(true), // Flow EC

            new SimpleLockThreadPool(false), // Don't flow EC

        };



            long[] queueCost = new long[threadPools.Length];

            long[] drainCost = new long[threadPools.Length];



            Console.WriteLine("+ Running benchmarks ({0}) +", threadPools.Length);



            for (int i = 0; i < threadPools.Length; i++)
            {

                IThreadPool itp = threadPools[i];

                Console.Write("#{0} {1}: ", i, itp.ToString().PadRight(26));



                // Warm up:

                using (CountdownEvent cev = new CountdownEvent(warmupRunsPerThreadPool))
                {

                    WaitCallback wc = delegate { cev.Decrement(); };

                    for (int j = 0; j < warmupRunsPerThreadPool; j++)

                        itp.QueueUserWorkItem(wc, null);

                    cev.Wait();

                }



                // Now do the real thing:

                int g0collects = GC.CollectionCount(0);

                int g1collects = GC.CollectionCount(1);

                int g2collects = GC.CollectionCount(2);



                using (CountdownEvent cev = new CountdownEvent(realRunsPerThreadPool))

                using (ManualResetEvent gun = new ManualResetEvent(false))
                {

                    WaitCallback wc = delegate
                    {

                        if (separateQueueFromDrain) { gun.WaitOne(); }

                        cev.Decrement();

                    };

                    Stopwatch sw = Stopwatch.StartNew();

                    for (int j = 0; j < realRunsPerThreadPool; j++)

                        itp.QueueUserWorkItem(wc, null);

                    queueCost[i] = sw.ElapsedTicks;

                    sw = Stopwatch.StartNew();

                    if (separateQueueFromDrain) { gun.Set(); }

                    cev.Wait();

                    drainCost[i] = sw.ElapsedTicks;

                }



                g0collects = GC.CollectionCount(0) - g0collects;

                g1collects = GC.CollectionCount(1) - g1collects;

                g2collects = GC.CollectionCount(2) - g2collects;



                Console.WriteLine("q: {0}, d: {1}, t: {2} (collects: 0={3},1={4},2={5})",

                    queueCost[i].ToString("#,##0"),

                    drainCost[i].ToString("#,##0"),

                    (queueCost[i] + drainCost[i]).ToString("#,##0"),

                    g0collects,

                    g1collects,

                    g2collects

                );



                itp.Dispose();

                GC.Collect(2);

                GC.WaitForPendingFinalizers();

            }



            Console.WriteLine();

            Console.WriteLine("+ Comparison against baseline ({0}) +", threadPools[0]);

            for (int i = 0; i < threadPools.Length; i++)
            {

                Console.WriteLine("#{0} {1}: q: {2}x, d: {3}x, t: {4}x",

                    i,

                    threadPools[i].ToString().PadRight(26),

                    queueCost[i] / (float)queueCost[0],

                    drainCost[i] / (float)drainCost[0],

                    (queueCost[i] + drainCost[i]) / ((float)queueCost[0] + drainCost[0])

                );

            }

        }

    }
*/
}
